package com.vcg.neo4j.demo.rxjava;

import org.json.XMLTokener;
import org.junit.Test;
import org.springframework.cloud.netflix.rx.RxResponse;
import org.springframework.http.ResponseEntity;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.web.client.AsyncRestTemplate;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitter;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import rx.Observable;
import rx.Subscriber;
import rx.functions.Func1;
import rx.functions.Func2;
import rx.observables.GroupedObservable;

import java.util.Arrays;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

import static javafx.scene.input.KeyCode.O;
import static org.json.XMLTokener.entity;

/**
 * Created by wuyu on 2016/7/30.
 */
public class ObserversTest {


    @Test
    public void testResponseBodyEmitterSubscriber() throws ExecutionException, InterruptedException {
        AsyncRestTemplate restTemplate= new AsyncRestTemplate();
        ListenableFuture<ResponseEntity<String>> future = restTemplate.getForEntity("http://localhost:8080/ob/messages", String.class);
        ResponseEntity<String> entity1 = future.get();
        System.err.println(entity1.getBody());
    }

    @Test
    public void testReduce(){
        Observable.just(1,2,3,4,5)
                .reduce((integer, integer2) -> integer+integer2)
                .subscribe(System.err::println);
    }

   @Test
    public void testMerge(){
       Observable<Integer> ob1 = Observable.just(1, 2, 3, 4, 5);
       Observable<Integer> ob2 = Observable.just(67, 8, 9);
       Observable.merge(ob1,ob2)
               .subscribe(System.err::println);
   }


    @Test
    public void testGroupBy() {
        Observable.just(1, 2, 3, 4)
                .groupBy(integer -> integer % 2)
                .subscribe(k-> System.err.println(k.getKey()));
    }


    @Test
    public void testSample() {
        Observable.just(1, 2, 3, 3, 3, 12, 3, 1, 2, 1)
                .sample(1, TimeUnit.MILLISECONDS)
                .subscribe(System.err::println);
    }

    @Test
    public void testIgnoreElements() {
        Observable.just(1, 2, 3)
                .ignoreElements()
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onCompleted() {
                        System.err.println("完成");
                    }

                    @Override
                    public void onError(Throwable e) {
                        System.err.println(e.getMessage());
                    }

                    @Override
                    public void onNext(Integer integer) {
                        System.err.println(integer);
                    }
                });
    }

    @Test
    public void testFirst() {
        Observable.just(1, 2, 3, 4)
                .first()
                .subscribe(System.err::println);
    }

    @Test
    public void testFilter() {
        Observable.just(1, 2, 3, 4, 5)
                .filter(integer -> integer % 2 == 0)
                .subscribe(System.err::println);
    }


    @Test
    public void testFlatMap() {
        Observable.just(1, 2, 3, 4)
                .flatMap((Func1<Integer, Observable<?>>) integer -> Observable.just(integer + 1))
                .subscribe(System.err::println);
    }


    @Test
    public void testDistinct() {
        Observable.just(1, 2, 3, 1, 1, 1, 1, 1, 3, 2)
                .distinct()
                .subscribe(System.err::println);
    }

    @Test
    public void testScan() {
        Observable.just(1, 2, 3, 4, 5, 6)
                .scan((arg1, arg2) -> arg1 + arg2)
                .subscribe(System.err::println);
    }

    @Test
    public void testFrom() {
        Observable.from(Arrays.asList(1, 2, 3, 4, 5, 6)).subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(Integer integer) {
                System.err.println(integer);
            }
        });
    }

    @Test
    public void testCreate() {
        Observable<String> observable = Observable.create(subscriber -> {
            subscriber.onNext("消息1");
            subscriber.onNext("消息2");
            subscriber.onCompleted();
        });
        observable.subscribe(System.err::println);
    }

    @Test
    public void testJust() {
        Observable<Integer> observable = Observable.just(1, 2, 3, 4, 5, 6, 7, 8);
        observable.subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {
                System.err.println("发送完成");
            }

            @Override
            public void onError(Throwable e) {
                System.err.println("发送失败:" + e.getMessage());
            }

            @Override
            public void onNext(Integer integer) {
                System.err.println("消费消息:" + integer);
            }
        });
    }
}
